package com.lagou.kafka.demo.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class MyProducer1 {

    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        Map<String,Object> configs = new HashMap<String, Object>();
        // 设置连接Kafka的初始连接⽤到的服务器地址
        // 如果是集群，则可以通过此初始连接发现集群中的其他broker
        configs.put("bootstrap.servers","hadoop001:9092");
        // 设置key的序列化器
        configs.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        // 设置value的序列化器
        configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        configs.put("acks", "1");
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(configs);
        // ⽤于封装Producer的消息
        ProducerRecord<Integer, String> topic_1 = new ProducerRecord<Integer, String>(
                "topic_1",
                0, // 分区编号，现在只有⼀个分区，所以是0
                0, // 数字作为key
                "message 01" // 字符串作为value
        );
        // 发送消息，同步等待消息的确认
        producer.send(topic_1).get(3000, TimeUnit.MILLISECONDS);
        // 关闭⽣产者
        producer.close();
    }
}
